package com.raymond.queue.client.thread;

import com.raymond.queue.client.QueueClient;
import com.raymond.queue.client.msg.MessageManages;
import com.raymond.queue.client.queue.PackBlockingQueue;
import com.raymond.queue.common.msg.BaseMessage;
import com.raymond.queue.common.msg.request.PackRequest;
import com.raymond.queue.common.thread.BaseCirculationThread;
import com.raymond.queue.common.utils.CachedMillisecondClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 打包发送线程
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-07-19 11:39
 */
public class PackRequestThread extends BaseCirculationThread {
    private final static Logger logger = LoggerFactory.getLogger(PackRequestThread.class);
    private final PackBlockingQueue packTask;

    private final QueueClient queueClient;

    private final int packCount;

    private final BlockingQueue<MessageManages> asyncTask;

    public PackRequestThread(PackBlockingQueue packTask, BlockingQueue<MessageManages> asyncTask, QueueClient queueClient, int packCount) {
        this.packTask = packTask;
        this.asyncTask = asyncTask;
        this.queueClient = queueClient;
        this.packCount = packCount;
    }


    @Override
    protected void execute() {
        try {
            if (!queueClient.isDone()) {
                queueClient.sync();
            }
            if (packTask.size() < packCount) {
                packTask.wait0(10);
            }
            List<BaseMessage> poll = packTask.poll(packCount);
            if (poll == null || poll.size() == 0) {
                return;
            }
            ConcurrentHashMap<Integer, MessageManages> msgManager = queueClient.getMsgManager();
            queueClient.writeAndFlush(new PackRequest(poll)).addListener(future -> {
                if (future.isSuccess()) {
                    poll.forEach(req -> asyncTask.add(msgManager.get(req.getSequenceId()).setStartTime(CachedMillisecondClock.INS.now())));
                } else {
                    poll.forEach(req -> msgManager.get(req.getSequenceId()).asyncSendFail());
                }
            });;
            packTask.notifyAll0();
        } catch (Exception e) {
            logger.error("打包发送线程异常:", e);
        }
    }

}
